package com.ls.fw.data.hbase.support;

import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.hadoop.hbase.client.Append;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Increment;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Row;

import com.ls.fw.data.hbase.support.task.CallbackTask;
import com.ls.fw.data.hbase.support.task.ResultTask;
import com.ls.fw.data.hbase.support.task.Task;

/**
 * 
 * @author ls
 */
public class HAsyncTable {

	private LinkedBlockingQueue<AsyncPackage> asyncQueue;

	public HAsyncTable(LinkedBlockingQueue<AsyncPackage> asyncQueue) {
		this.asyncQueue = asyncQueue;
	}

	public void get(Get get, ResultTask callback) {
		asyncQueue.offer(new AsyncPackage(get, callback));
	}

	public void append(Append append, ResultTask callback) {
		asyncQueue.offer(new AsyncPackage(append, callback));
	}

	public void increment(Increment increment, ResultTask callback) {
		asyncQueue.offer(new AsyncPackage(increment, callback));
	}

	public void put(Put put, CallbackTask callback) {
		asyncQueue.offer(new AsyncPackage(put, callback));
	}

	public void delete(Delete delete, CallbackTask callback) {
		asyncQueue.offer(new AsyncPackage(delete, callback));
	}

	public void batch(List<Row> rows, ResultTask task) {
		rows.parallelStream()
			.forEach(row -> asyncQueue.offer(new AsyncPackage(row, task)));
	}

	public class AsyncPackage {
		public Row action;
		public Task callback;

		public AsyncPackage(Row action, Task callback) {
			this.action = action;
			this.callback = callback;
		}
	}
}
